package com.class1group3.demo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;

@Component
public class HotPublishersInitializer implements CommandLineRunner {

    private static final Logger log = LoggerFactory.getLogger(HotPublishersInitializer.class);

    private final JdbcTemplate jdbcTemplate;

    @Autowired
    public HotPublishersInitializer(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public void run(String... args) {
        try {
            log.info("开始初始化热门出版社统计表...");

            // 1. 设置Hive优化参数
            setHiveOptimizationParams();

            // 2. 创建热门出版社统计表
            createHotPublishersTable();

            // 3. 执行数据插入
            insertHotPublishersData();

            log.info("热门出版社统计表初始化成功");
        } catch (DataAccessException e) {
            handleInitializationError(e);
        }
    }

    private void setHiveOptimizationParams() {
        try {
            jdbcTemplate.execute("SET hive.exec.dynamic.partition=true");
            jdbcTemplate.execute("SET hive.exec.dynamic.partition.mode=nonstrict");
            jdbcTemplate.execute("SET hive.vectorized.execution.enabled=true");
            jdbcTemplate.execute("SET hive.map.aggr=true");
            log.info("Hive优化参数设置成功");
        } catch (DataAccessException e) {
            log.warn("设置Hive优化参数失败: {}", e.getMessage());
        }
    }

    private void createHotPublishersTable() {
        String createTableSql = "CREATE TABLE IF NOT EXISTS hot_publishers (\n" +
                "    publisher STRING COMMENT '出版社名称',\n" +
                "    book_count BIGINT COMMENT '书籍数量',\n" +
                "    rank INT COMMENT '排名'\n" +
                ")\n" +
                "COMMENT '热门出版社统计表（TOP 20）'\n" +
                "STORED AS ORC\n" +
                "TBLPROPERTIES (\n" +
                "    'orc.compress'='SNAPPY',\n" +
                "    'transactional'='false'\n" +
                ")";

        try {
            jdbcTemplate.execute(createTableSql);
            log.info("热门出版社统计表创建成功");
        } catch (DataAccessException e) {
            log.error("创建表失败: {}", e.getMessage());
            handleTableCreationError();
        }
    }

    private void handleTableCreationError() {
        log.error("表创建失败，尝试备选方案...");

        // 备选方案1：尝试删除后重建
        try {
            jdbcTemplate.execute("DROP TABLE IF EXISTS hot_publishers");
            createHotPublishersTable();
            log.info("通过删除重建成功创建表");
        } catch (DataAccessException e) {
            log.error("删除重建失败: {}", e.getMessage());

            // 备选方案2：使用简化表结构
            createSimpleTable();
        }
    }

    private void createSimpleTable() {
        String simpleCreateSql = "CREATE TABLE IF NOT EXISTS hot_publishers (\n" +
                "    publisher STRING,\n" +
                "    book_count BIGINT,\n" +
                "    rank INT\n" +
                ") STORED AS ORC";

        try {
            jdbcTemplate.execute(simpleCreateSql);
            log.info("简化版热门出版社统计表创建成功");
        } catch (DataAccessException e) {
            log.error("简化版表创建失败: {}", e.getMessage());
        }
    }

    private void insertHotPublishersData() {
        String insertSql = "INSERT OVERWRITE TABLE hot_publishers\n" +
                "SELECT\n" +
                "    publisher,\n" +
                "    book_count,\n" +
                "    ROW_NUMBER() OVER (ORDER BY book_count DESC) AS rank\n" +
                "FROM (\n" +
                "    SELECT\n" +
                "        COALESCE(publisher, '未知出版社') AS publisher,\n" +
                "        COUNT(*) AS book_count\n" +
                "    FROM books\n" +
                "    GROUP BY publisher\n" +
                "    ORDER BY book_count DESC\n" +
                "    LIMIT 20\n" +
                ") t";

        try {
            jdbcTemplate.execute(insertSql);
            log.info("热门出版社数据插入成功");

            // 验证结果
            Integer count = jdbcTemplate.queryForObject(
                    "SELECT COUNT(*) FROM hot_publishers", Integer.class);
            log.info("热门出版社表记录数: {}", count);
        } catch (DataAccessException e) {
            log.error("数据插入失败: {}", e.getMessage());
            handleInsertError();
        }
    }

    private void handleInsertError() {
        log.error("数据插入失败，尝试备选方案...");

        // 备选方案1：使用临时表
        try {
            useTemporaryTableMethod();
        } catch (DataAccessException e) {
            log.error("临时表方法失败: {}", e.getMessage());

            // 备选方案2：使用CTAS方法
            useCTASMethod();
        }
    }

    private void useTemporaryTableMethod() {
        log.info("尝试临时表分步计算方案...");

        // 步骤1：创建临时表存储基础数据
        jdbcTemplate.execute("CREATE TEMPORARY TABLE temp_publisher_counts STORED AS ORC AS\n" +
                "SELECT\n" +
                "    COALESCE(publisher, '未知出版社') AS publisher,\n" +
                "    COUNT(*) AS book_count\n" +
                "FROM books\n" +
                "GROUP BY publisher\n" +
                "ORDER BY book_count DESC\n" +
                "LIMIT 20");

        // 步骤2：计算排名并插入目标表
        String insertFromTemp = "INSERT OVERWRITE TABLE hot_publishers\n" +
                "SELECT\n" +
                "    publisher,\n" +
                "    book_count,\n" +
                "    ROW_NUMBER() OVER (ORDER BY book_count DESC) AS rank\n" +
                "FROM temp_publisher_counts";

        jdbcTemplate.execute(insertFromTemp);

        // 清理临时表
        jdbcTemplate.execute("DROP TABLE temp_publisher_counts");

        log.info("通过临时表分步计算成功插入数据");
    }

    private void useCTASMethod() {
        log.info("尝试CTAS方案...");

        // 创建新表替换旧表
        String ctasSql = "CREATE TABLE hot_publishers_new STORED AS ORC AS\n" +
                "SELECT\n" +
                "    publisher,\n" +
                "    book_count,\n" +
                "    ROW_NUMBER() OVER (ORDER BY book_count DESC) AS rank\n" +
                "FROM (\n" +
                "    SELECT\n" +
                "        COALESCE(publisher, '未知出版社') AS publisher,\n" +
                "        COUNT(*) AS book_count\n" +
                "    FROM books\n" +
                "    GROUP BY publisher\n" +
                "    ORDER BY book_count DESC\n" +
                "    LIMIT 20\n" +
                ") t";

        try {
            // 创建新表
            jdbcTemplate.execute(ctasSql);

            // 重命名表
            jdbcTemplate.execute("ALTER TABLE hot_publishers RENAME TO hot_publishers_old");
            jdbcTemplate.execute("ALTER TABLE hot_publishers_new RENAME TO hot_publishers");
            jdbcTemplate.execute("DROP TABLE hot_publishers_old");

            log.info("通过CTAS成功创建并替换表");
        } catch (DataAccessException e) {
            log.error("CTAS方案失败: {}", e.getMessage());
        }
    }

    private void handleInitializationError(DataAccessException e) {
        log.error("热门出版社统计表初始化失败", e);

        if (e.getCause() != null) {
            String errorMsg = e.getCause().getMessage();

            if (errorMsg.contains("Permission denied")) {
                log.error("=== HDFS权限问题 ===");
                log.error("解决方案：");
                log.error("1. 在HiveServer2节点执行: hdfs dfs -chmod -R 777 /user/hive/warehouse");
                log.error("2. 或联系Hadoop管理员添加写权限");
            }
            else {
                log.error("=== 其他错误 ===");
                log.error("错误详情: {}", errorMsg);
            }
        }
    }
}
